package clickhouse_tb

import (
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"wx-api/adx-bid/config"
)

func CkRunFlush() {
	AdxMsgCk.Flush()
	SspMsgCk.Flush()
	DspMsgCk.Flush()
}

// ck
type CkADB struct {
	Table        string
	SqlStr       string
	Fields       []string
	BulkLen      int
	FlushTime    int
	Mux          *sync.Mutex
	Execution_id uint64
	Events       [][]interface{}
}

func (j *CkADB) SqlStrInit() {
	fieldsv := make([]string, len(j.Fields))
	for key, _ := range j.Fields {
		fieldsv[key] = "?"
	}
	query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", j.Table, strings.Join(j.Fields, ","), strings.Join(fieldsv, ","))
	j.SqlStr = query
}

func (j *CkADB) InsertClk() {
	db := config.GetCKDB()
	if db == nil {
		for i := 0; i < 10; i++ {
			time.Sleep(100 * time.Millisecond)
			db = config.GetCKDB()
			if db != nil {
				break
			}
		}
		if db == nil {
			fmt.Println("clickhouse db is nil")
			return
		}
	}
	tx, err := db.Begin()
	if err != nil {
		fmt.Println("tx clickhouse is error:" + err.Error())
		return
	}
	stmt, err := tx.Prepare(j.SqlStr)
	if err != nil {
		fmt.Println("tx clickhouse is error:" + err.Error())
	}
	defer tx.Rollback()
	defer stmt.Close()
	er_num := 0
	for _, event := range j.Events {
		if len(event) == 0 {
			continue
		}
		if _, err := stmt.Exec(event...); err != nil {
			fmt.Println("exec clickhouse insert error:", err.Error())
			er_num++
			continue
		}
	}
	if err := tx.Commit(); err != nil {
		fmt.Println("exec clickhouse commit is error:", err.Error())
		return
	}
	execution_id := atomic.AddUint64(&j.Execution_id, 1)
	fmt.Println("write ", j.Table, (len(j.Events) - er_num), " docs to clickhouse with execution_id ", execution_id)
	return
}

func (j *CkADB) Flush() {
	go func() {
		for range time.NewTicker(time.Second * time.Duration(j.FlushTime)).C {
			j.Mux.Lock()
			if j.SqlStr == "" { //初始化sql
				j.SqlStrInit()
			}
			if len(j.Events) > 0 {
				j.InsertClk()
				j.Events = nil
			}
			j.Mux.Unlock()
		}
	}()
}

func (j *CkADB) Emit(event []interface{}) {
	j.Mux.Lock()
	defer j.Mux.Unlock()
	if j.SqlStr == "" {
		j.SqlStrInit()
	}
	j.Events = append(j.Events, event)
	if len(j.Events) > j.BulkLen {
		j.InsertClk()
		j.Events = nil
	}
}
